package cn.qdhlm.frontendprocessor.core.callback;


import cn.qdhlm.frontendprocessor.common.helper.RocketMqProducer;
import cn.qdhlm.frontendprocessor.common.util.ComUtil;
import cn.qdhlm.frontendprocessor.common.util.JsonUtil;
import cn.qdhlm.frontendprocessor.model.memory.CollectorSession;
import cn.qdhlm.frontendprocessor.model.memory.CollectorSessionMap;
import cn.qdhlm.frontendprocessor.model.memory.SocketSession;
import cn.qdhlm.frontendprocessor.model.memory.SocketSessionMap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.concurrent.ScheduledFuture;

/**
 * @Name: SocketChannelInHandler
 * @Description: SocketChannelInHandler
 * @Author: jiangcs
 */
@ChannelHandler.Sharable
@Log4j2
@Component
public class SocketChannelInHandler extends ChannelInboundHandlerAdapter {
    @Autowired
    private RocketMqProducer producer;

    /**
     * ChannelUnregistered：刚创建;
     * ChannelRegistered: (创建以后),已注册到EventLoop;
     * ChannelActive: 已经连接到远程节点;
     * ChannelInactive: 没有连接到远程节点。
     */

    @Override
    public void channelActive(ChannelHandlerContext chc) {
        String socketId = getSocketId(chc);

        log.info("channel active: socketId[{}]", socketId);

        // 保存SocketSession
        SocketSession socketSession = new SocketSession();
        socketSession.setChannelHandlerContext(chc);
        SocketSessionMap.add(socketId, socketSession);
    }

    @Override
    public void channelRead(ChannelHandlerContext chc, Object msg) {
        String socketId = getSocketId(chc);

        // 数据ByteBuf
        String hexStr = ByteBufUtil.hexDump((ByteBuf) msg);

        log.info("channel read: socketId[{}] => {}", socketId, hexStr);

        byte[] bytes = ByteBufUtil.getBytes((ByteBuf)msg);


        // 发送消息
        System.out.println("byte[]::::"+new String(bytes));
        producer.send("DeviceData", "socket", ByteBufUtil.getBytes((ByteBuf)msg));
    }

    @Override
    public void channelInactive(ChannelHandlerContext chc) {
        String socketId = getSocketId(chc);

        log.warn("channel inactive: socketId[{}]", socketId);

//        cleanSessionAndScheduleTask(socketId);

        chc.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext chc, Throwable cause) {
        String socketId = getSocketId(chc);

        log.error("exception caught: socketId[{}] => ", socketId, cause);

//        cleanSessionAndScheduleTask(socketId);

        chc.close();
    }

    /**
     * 根据 ip、port 获取 socketId
     *
     * @param chc
     * @return
     */
    private String getSocketId(ChannelHandlerContext chc) {
        String socketId = "unknown";
        InetSocketAddress inetSocketAddress = (InetSocketAddress) chc.channel().remoteAddress();
        if (inetSocketAddress != null) {
            socketId = inetSocketAddress.getAddress().getHostAddress() + "_" + inetSocketAddress.getPort();
        }
        return socketId;
    }

    /**
     * 清理 session 和 scheduleTask
     *
     * @param socketId
     */
    private void cleanSessionAndScheduleTask(String socketId) {
        // 移除 socketSession
        SocketSession socketSession = SocketSessionMap.remove(socketId);
        if (!ComUtil.isEmpty(socketSession)) {
            String collectorNo = socketSession.getCollectorNo();
            if (null != collectorNo) {
                // 移除collectorSession
                CollectorSession collectorSession = CollectorSessionMap.remove(collectorNo);

                // 采集器在线记录：离线
//                Integer onlineId = onlineCollectorMapper.getOnlineIdByCollectorNo(collectorNo);
//                if (!ComUtil.isEmpty(onlineId)) {
//                    int count = onlineCollectorMapper.updateOffline(onlineId);
//
//                    log.info("update collector[{}] offline: {}", collectorNo, count);
//                }

                // 取消定时任务
                if (!ComUtil.isEmpty(collectorSession)) {
                    for (ScheduledFuture<?> scheduledFuture : collectorSession.getScheduledFutureList()) {
                        scheduledFuture.cancel(true);

                        log.info("cancel schedule task: {}", JsonUtil.safeJsonStr(scheduledFuture));
                    }
                }

                log.warn("clean socketId[{}] collectorNo[{}] => socketSession[{}] collectorSession[{}]", socketId, collectorNo, JsonUtil.safeJsonStr(socketSession), JsonUtil.safeJsonStr(collectorSession));
            } else {
                log.warn("clean socketId[{}] => socketSession[{}]", socketId, JsonUtil.safeJsonStr(socketSession));
            }
        }
    }
}

